package job

import (
	"context"
	comet "fgo/advance/other/discovery/bilibili/example/api/coment/grpc"
	"fgo/advance/other/discovery/bilibili/example/conf"
	"fmt"
	"github.com/bilibili/discovery/naming"
	"github.com/siddontang/go/log"
	"net/url"
	"time"
)

// Job is push job.
type Job struct {
	c            *conf.Config
	cometServers map[string]*Comet
}

// New new a push job.
func New(c *conf.Config) *Job {
	j := &Job{
		c: conf.Conf,
	}
	j.watchComet(c.Discovery)
	return j
}

// 模拟发送数据
func (j *Job) Send() {
	for {
		for _, v := range j.cometServers {
			resp, err := v.client.SayHello(context.Background(), &comet.HelloRequest{
				Name: "Job",
			})
			fmt.Println(resp, err)
		}
		time.Sleep(time.Second * 5)
	}
}

func (j *Job) watchComet(c *naming.Config) {
	dis := naming.New(c)
	resolver2 := dis.Build("goim.test")
	event := resolver2.Watch()
	select {
	case _, ok := <-event:
		if !ok {
			panic("watchComet init failed")
		}
		if ins, ok := resolver2.Fetch(); ok {
			if err := j.newAddress(ins.Instances); err != nil {
				panic(err)
			}
			log.Infof("watchComet init newAddress:%+v", ins)
		}
	case <-time.After(10 * time.Second):
		log.Error("watchComet init instances timeout")
	}
	go func() {
		for {
			if _, ok := <-event; !ok {
				log.Info("watchComet exit")
				return
			}
			ins, ok := resolver2.Fetch()
			if ok {
				if err := j.newAddress(ins.Instances); err != nil {
					log.Errorf("watchComet newAddress(%+v) error(%+v)", ins, err)
					continue
				}
				log.Infof("watchComet change newAddress:%+v", ins)
			}
		}
	}()
}

// NewComet new a comet.
func (j *Job) newAddress(insMap map[string][]*naming.Instance) error {
	ins := insMap[j.c.Env.Zone]
	if len(ins) == 0 {
		return fmt.Errorf("watchComet instance is empty")
	}
	comets := map[string]*Comet{}
	for _, in := range ins {
		if old, ok := j.cometServers[in.Hostname]; ok {
			comets[in.Hostname] = old
			continue
		}
		//c, err := NewComet(in, j.c.Comet)
		c, err := NewComet(in)
		if err != nil {
			log.Errorf("watchComet NewComet(%+v) error(%v)", in, err)
			return err
		}
		comets[in.Hostname] = c
		log.Infof("watchComet AddComet grpc:%+v", in)
	}
	for key, old := range j.cometServers {
		if _, ok := comets[key]; !ok {
			old.cancel()
			log.Infof("watchComet DelComet:%s", key)
		}
	}
	j.cometServers = comets
	return nil
}

// NewComet new a comet.
func NewComet(in *naming.Instance) (*Comet, error) {
	cmt := &Comet{
		serverID: in.Hostname,
	}

	var grpcAddr string
	// 方式一: 使用 appid 通信
	//grpcAddr = "discovery://default/goim.test"
	// 方式二: 遍历 addrs,获取对应的服务地址
	for _, addrs := range in.Addrs {
		u, err := url.Parse(addrs)
		if err == nil && u.Scheme == "grpc" {
			grpcAddr = u.Host
		}
	}
	if grpcAddr == "" {
		return nil, fmt.Errorf("invalid grpc address:%v", in.Addrs)
	}

	var err error
	if cmt.client, err = newCometClient(grpcAddr); err != nil {
		return nil, err
	}
	cmt.ctx, cmt.cancel = context.WithCancel(context.Background())
	return cmt, nil
}
